package com.gugee.ins.data.collector.bloggerlocation.service.impl;

import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.ins.common.constant.InsBloggerMsgTypeCodeEnum;
import com.gugee.ins.common.dal.service.blogger.*;
import com.gugee.ins.common.dal.spider.InsBloggerSpiderClient;
import com.gugee.ins.common.dal.spider.InsMediaSpiderClient;
import com.gugee.ins.common.model.mysql.blogger.*;
import com.gugee.ins.common.model.spider.app.media.InsAppMediaBody;
import com.gugee.ins.common.model.spider.app.media.InsAppMediaResponse;
import com.gugee.ins.common.model.spider.blogger.*;
import com.gugee.ins.data.collector.bloggerlocation.config.InsBloggerLocationConfig;
import com.gugee.ins.data.collector.bloggerlocation.service.InsBloggerLocationService;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ForkJoinPool;

/**
 * insBloggerLocation 服务
 * @author Huangxin
 */
@Slf4j
@Service
public class InsBloggerLocationServiceImpl implements InsBloggerLocationService {

    /**
     * 配置
     */
    @Autowired
    InsBloggerLocationConfig insBloggerLocationConfig;

    /**
     * 线程调度
     */
    @Autowired
    ScheduleTemplate scheduleTemplate;

    /**
     * kafkaTemplate
     */
    @Autowired
    KafkaTemplate<Long,InsBloggerMessage> kafkaTemplateInsBlogger;

    /**
     * 爬虫接口
     */
    @Autowired
    InsBloggerSpiderClient insBloggerSpiderClient;

    /**
     * 爬虫接口
     */
    @Autowired
    InsMediaSpiderClient insMediaSpiderClient;

    /**
     * InsBloggerDao
     */
    @Autowired
    InsBloggerDao insBloggerDao;

    /**
     * InsBloggerDao
     */
    @Autowired
    InsBloggerMainDao insBloggerMainDao;

    /**
     *
     */
    @Autowired
    InsRelBloggerLocationDao insRelBloggerLocationDao;

    /**
     * 线程池
     */
    ForkJoinPool forkJoinPool;

    /**
     * 线程名
     */
    private String workName = "insBloggerLocation-worker";

    /**
     * 初始化线程池
     */
    @PostConstruct
    public void init() {
        forkJoinPool = ThreadPoolUtils.forkJoinPool(insBloggerLocationConfig.getInsBloggerFollowingCorePollSize(), workName);
        scheduleTemplate.attach(workName, forkJoinPool);
    }

    /**
     * 处理 博主地理位置 更新 任务
     * @param insBloggerMessages
     */
    @Override
    public void handleLocationCollect(List<InsBloggerMessage> insBloggerMessages) {
        try {
            forkJoinPool.submit(() -> insBloggerMessages.parallelStream().forEach(msg -> {
                try {
                    // 采集播主关注人数据
                    runCollectLocation(msg);
                } catch (Exception ex) {
                    ex.printStackTrace();
                    log.error("location collect error. uid:{}, error :{}" ,
                            msg.getUid(),ex.getMessage());
                }
            })).get();
        } catch (Exception ex) {
            log.error("location collect thread error.  {}" ,  ex.getMessage());
        }
    }

    /**
     * 采集 博主地理位置
     * 先爬取 detail接口，如果没有再爬取帖子接口
     * @param insBloggerMessage
     */
    public void runCollectLocation(InsBloggerMessage insBloggerMessage){
        long start = System.currentTimeMillis();
        long uid = insBloggerMessage.getUid();
        // 时间范围
        long limitTime = start / 1000 - insBloggerLocationConfig.getInsBloggerLocationClIntervalDay() * 24 * 3600;

        InsBlogger blogger = insBloggerDao.findByUid(uid);
        if(blogger == null){
            return;
        }

        // 判断是否需要采集地理位置
        if(insBloggerMessage.getTypeCode() != InsBloggerMsgTypeCodeEnum.NO_HANDLE_AND_DO_OTHER_OPT.getCode()){
            try{
                runCollectLocation(uid,true,0,blogger.getLastLocationCollectTime());
            }catch (Exception e){
                log.error(" insMainBlogger location collect error. uid:{}, error :{}",uid,e.getMessage());
            }
        }

        // 查询他的活跃粉丝
        List<InsBlogger> subMessages = insBloggerMainDao.findInsLocationCollectMessageByUid(insBloggerMessage.getUid(),limitTime);
        if(subMessages != null && !subMessages.isEmpty()){
            subMessages.forEach(msg -> {
                try{
                    runCollectLocation(msg.getUid(), msg.getFollowerCount() >= insBloggerLocationConfig.getInsBloggerMainCollectFollowerCount(),
                            msg.getMediaCount(),msg.getLastLocationCollectTime());
                }catch (Exception e){
                    log.error("insBlogger location collect error. uid:{}, error :{}",msg.getUid(),e.getMessage());
                }

            });
        }

        sendPriorityAnalysisLocationMsg(insBloggerMessage,blogger);

        if ((System.currentTimeMillis() - start) / 1000 > insBloggerLocationConfig.getBloggerLocationDebugTimeLimit()) {
            log.info("ins Blogger Location Collect uid:{} use time:{}s", insBloggerMessage.getUid(),(System.currentTimeMillis() - start) / 1000);
        }
    }

    /**
     * 采集 博主地理位置
     * 先爬取 detail接口，如果没有再爬取帖子接口
     * @param uid
     * @param isMainBlogger
     * @param mediaCount
     * @param lastLocationCollectTime
     */
    public void runCollectLocation(long uid,boolean isMainBlogger,int mediaCount,long lastLocationCollectTime){
        String cityName = "";
        String pageInfo;
        int[] collectMediaCount = {0};
        List<InsRelBloggerLocation> insRelBloggerLocationList = new ArrayList<>(insBloggerLocationConfig.getInsCollectMediaMaxCount());

        // 根据uid爬取detail信息
        try {
            InsBloggerDetailResponse insBloggerDetailResponse = insBloggerSpiderClient.getInsBloggerDetail(uid);
            if(insBloggerDetailResponse.getPrimaryCountryInfo() != null){
                cityName = insBloggerDetailResponse.getPrimaryCountryInfo().getCountryName();
            }
        }catch (Exception e){
            log.error("spider ins blogger detail error,msg:{}",e.getMessage());
        }

        // 如果还是空的就去爬取帖子
        if(StringUtils.isEmpty(cityName)){
            // 大博主就不再采集帖子直接从库表读取
            if(!isMainBlogger){
                if(mediaCount > 0){
                    pageInfo = spiderMedia(uid,"",collectMediaCount,insRelBloggerLocationList,lastLocationCollectTime);
                    while(!StringUtils.isEmpty(pageInfo)
                            && insBloggerLocationConfig.getInsCollectMediaMaxCount() > collectMediaCount[0]){
                        pageInfo = spiderMedia(uid,pageInfo,collectMediaCount,insRelBloggerLocationList,lastLocationCollectTime);
                    }
                    // 批量保存
                    insRelBloggerLocationDao.batchInsert(insRelBloggerLocationList);
                }
            }else{
                // 库表读取帖子 location 位置后直接插入关系表
                if(lastLocationCollectTime >= 0){
                    insRelBloggerLocationDao.insertByUidTimeLimit(uid,lastLocationCollectTime);
                }
            }
        } else {
            // 保存 detail 中爬到的地理位置信息
            String region = cityName.substring(cityName.lastIndexOf(",") + 1).trim();
            try{
                InsBlogger insBlogger = new InsBlogger();
                insBlogger.setUid(uid);
                insBlogger.setDetailCityName(cityName);
                insBlogger.setRegion(region);
                insBlogger.setLastLocationCollectTime(System.currentTimeMillis() / 1000);
                insBloggerDao.updateByUidSelective(insBlogger);
            }catch (Exception e){
                log.error("ins Blogger Location Collect save insBlogger error,msg:{}",e.getMessage());
            }
            try{
                InsBloggerMain insBloggerMain = new InsBloggerMain();
                insBloggerMain.setUid(uid);
                insBloggerMain.setDetailCityName(cityName);
                insBloggerMain.setRegion(region);
                insBloggerMainDao.updateByUidSelective(insBloggerMain);
            }catch (Exception e){
                log.error("ins Blogger Location Collect save insBloggerMain error,msg:{}",e.getMessage());
            }
        }
    }

    /**
     * 循环拉帖子
     * @param uid
     * @param count
     * @param endCursor
     * @param collectMediaCount
     * @param insRelBloggerLocationList
     * @param lastLocationCollectTime
     * @return
     */
    private String spiderMedia(long uid,String endCursor,int[] collectMediaCount,
                                 List<InsRelBloggerLocation> insRelBloggerLocationList,
                                 long lastLocationCollectTime){
        //根据uid爬取信息
        List<InsAppMediaBody> edges;
        InsAppMediaResponse insAppMediaResponse;
        String pageInfo;
        try {
            insAppMediaResponse = insMediaSpiderClient.getAppMediaByUid(uid,endCursor);
            edges = insAppMediaResponse.getItems();
            if(edges != null && !edges.isEmpty()){
                collectMediaCount[0] += edges.size();
                for(InsAppMediaBody edge : edges){
                    if(edge != null){
                        // 如果帖子发布时间比上次采集时间小就说明已经采集过了，后续的就不管了
                        if(edge.getTakenTime() < lastLocationCollectTime){
                            return null;
                        }
                        if(edge.getLocation() != null
                                && edge.getLocation().getId() != null
                                && !StringUtils.isEmpty(edge.getLocation().getName())){
                            InsRelBloggerLocation insLocation = new InsRelBloggerLocation();
                            insLocation.setLocationId(edge.getLocation().getId());
                            insLocation.setLocationName(edge.getLocation().getName());
                            insLocation.setCreateTime(System.currentTimeMillis() / 1000);
                            insLocation.setShortcode(edge.getShortcode());
                            insLocation.setUid(uid);
                            insRelBloggerLocationList.add(insLocation);
                        }
                    }
                }
            }
            pageInfo = insAppMediaResponse.getNextMaxId();
        }catch (Exception e){
            pageInfo = null;
            log.error("spider ins media for ins blogger detail error, msg :{}",e.getMessage());
        }
        return pageInfo;
    }

    /**
     * 发送优先分析地理位置的消息
     * @param insBloggerMessage
     * @param insBlogger
     */
    public void sendPriorityAnalysisLocationMsg(InsBloggerMessage insBloggerMessage,InsBlogger insBlogger){
        if(StringUtils.isEmpty(insBlogger.getNation())){
            insBloggerMessage.setTypeCode(InsBloggerMsgTypeCodeEnum.HANDLE_AND_DO_OTHER_OPT.getCode());
        }else{
            insBloggerMessage.setTypeCode(InsBloggerMsgTypeCodeEnum.NO_HANDLE_AND_DO_OTHER_OPT.getCode());
        }
        insBloggerMessage.setRetryTimes(insBloggerLocationConfig.getInsAnalysisLocationPriorityRetryTimes());
        insBloggerMessage.setRetryLimitTimes(insBloggerLocationConfig.getInsAnalysisLocationPriorityRetryTimes());
        insBloggerMessage.setEventTime(System.currentTimeMillis());
        kafkaTemplateInsBlogger.send(insBloggerLocationConfig.getInsAnalysisLocationPriorityTopic(),insBloggerMessage);
    }
}
